package mq

import (
	"context"
	"cyiris/core/protocol/gpb"
	log "cyiris/core/zaplog"

	"github.com/streadway/amqp"
	"google.golang.org/protobuf/proto"
)

type MqCfg map[string]string

type QueueCfg struct {
	MqName       string `yaml:"mq_name"`
	ExchangeName string `yaml:"exchange_name"`
	ExchangeType string `yaml:"exchange_type"` // "direct", "fanout", "topic" and "headers"
	RoutingKey   string `yaml:"routing_key"`
	QueueName    string `yaml:"queue_name"`
	Workers      int    `yaml:"workers"`
	MaxPriority  int    `yaml:"max_priority"`
}

type Task struct {
	MqCfg    MqCfg
	Conns    map[string]*Connector
	QueueCfg []QueueCfg
	CtrlMap  map[string]Ctrl
}

func NewTask(mqCfg MqCfg, queueCfg []QueueCfg) *Task {
	task := &Task{
		MqCfg:    mqCfg,
		QueueCfg: queueCfg,
	}

	task.Conns = make(map[string]*Connector)
	task.CtrlMap = make(map[string]Ctrl)

	task.initConnector()
	task.initConsumer()

	return task
}

func (task *Task) RegisterRouter(serviceName string, ctrl Ctrl) {
	task.CtrlMap[serviceName] = ctrl
}

func (task *Task) initConnector() {
	for name, addr := range task.MqCfg {
		task.Conns[name] = NewConnector(name, addr)
	}
}

func (task *Task) initConsumer() (err error) {
	processFunc := func(msg *amqp.Delivery, qName string) error {
		return task.processMsg(msg, qName)
	}

	for _, cfg := range task.QueueCfg {
		conn, ok := task.Conns[cfg.MqName]
		if !ok {
			log.Errorf("Connector<%s> not exixts ", cfg.MqName)
			continue
		}

		_, err = NewConsumer(conn, cfg.ExchangeName, cfg.ExchangeType, cfg.RoutingKey, cfg.QueueName, cfg.Workers, cfg.MaxPriority, processFunc)
		if nil != err {
			return err
		}
	}

	return nil
}

func (task *Task) processMsg(mqMsg *amqp.Delivery, qName string) error {
	msg := &gpb.Message{}
	err := proto.Unmarshal(mqMsg.Body, msg)
	if nil != err {
		log.Errorf("Unmarshal MQ Message failed! err: %s", err.Error())
	}

	log.Infof("processing task msg.Header: %#v, msg.Body", msg.Header, string(msg.Body))

	// 获取任务 ctrl
	ctrl, ok := task.CtrlMap[msg.Header.ServiceName]
	if !ok {
		log.Errorf("Task Service<%s> not found !", msg.Header.ServiceName)
	}

	ctx := NewContext(context.Background(), msg, mqMsg)

	defer func() {
		if err := recover(); err != nil {
			log.Errorf("processing task err: %#v", err)
		}
	}()
	task.ctrlAction(ctrl, ctx)

	return nil
}

func (task *Task) ctrlAction(ctrl Ctrl, ctx *Context) {
	// 执行 filter 定义的 DoBefore 函数
	filters := ctrl.Filters()
	for _, f := range filters {
		if err := f.DoBefore(ctx); err != nil {
			log.Warnf("Task Ctrl Before Filter Failed: %s", err)
			return
		}
	}

	// 执行控制器定义的前置函数
	if err := ctrl.DoBefore(ctx); err != nil {
		log.Warnf("Task Ctrl DoBefore Failed: %s", err)
		return
	}

	log.Debugf("exec task ctrl %T Action", ctrl)
	ctrl.Action(ctx)

	// 执行控制器定义的后置函数
	if err := ctrl.DoAfter(ctx); err != nil {
		log.Warnf("Task Ctrl DoAfter Failed: %s", err)
		return
	}

	// 执行 filter 定义的 Dobefor 函数，后注册的先执行
	length := len(filters) - 1
	for i := range filters {
		f := filters[length-i]
		if err := f.DoAfter(ctx); err != nil {
			log.Warnf("Task Ctrl After Filter Failed: %s", err)
			return
		}
	}
}
